import json
from contextlib import asynccontextmanager
from datetime import datetime, timedelta
import aioredis
import uvicorn
from fastapi import FastAPI, WebSocket
from binance import AsyncClient, BinanceSocketManager, BinanceWebsocketUnableToConnect
import asyncio
import config
import scheduler_config
from log_utils import AsyncLogger
from notify_utils import send_gotify

logger = AsyncLogger()
# 初始化调度器
scheduler = scheduler_config.setup_scheduler()
redis = aioredis.from_url(
    f"redis://:{config.REDIS_CONFIG['password']}@{config.REDIS_CONFIG['host']}:{config.REDIS_CONFIG['port']}/{config.REDIS_CONFIG['db']}"
    , encoding="utf-8", decode_responses=True
)
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 初始化异步客户端
    client = await AsyncClient.create(api_key=config.BINANCE_CONFIG['api_key'],
                                      api_secret=config.BINANCE_CONFIG['api_secret'],
                                      https_proxy=config.BINANCE_CONFIG['https_proxy'])
    # 添加ws监听任务
    scheduler.add_job(ws_all_symbol_price,
                      trigger='date',
                      run_date=datetime.now() + timedelta(seconds=5),
                      args=[client, redis],
                      id='ws_all_symbol_price_task',
                      max_instances=1,
                      coalesce=True,
                      misfire_grace_time=5
                      )
    scheduler.add_job(consume_crypto_stream,
                      trigger='date',
                      run_date=datetime.now() + timedelta(seconds=3),
                      args=[redis],
                      id='consume_crypto_stream_task',
                      max_instances=1,
                      coalesce=True,
                      misfire_grace_time=5
                      )
    scheduler.start()
    logger.info("初始化完成")
    yield
    await client.close_connection()
    await redis.close()
    logger.info('销毁完成')


app = FastAPI(lifespan=lifespan)


async def ws_all_symbol_price(client, redis_conn):
    bm = BinanceSocketManager(client)

    # 订阅实时数据
    async with bm.all_mark_price_socket(fast=False) as stream1, \
            bm.symbol_ticker_socket("ETHUSDT") as stream2:
        try:
            while True:
                try:
                    msg = await stream1.recv()
                    rid = msg['data'][0]['E']
                    symbol_dict = {}
                    for symbol in msg['data']:
                        symbol_dict[symbol['s']] = [symbol['p'], symbol['r']]
                    await redis_conn.xadd(config.REDIS_KEY_NAMES['all_price_stream'], {"data": json.dumps(symbol_dict)},
                                          id=rid,
                                          maxlen=100000,
                                          approximate=True)
                except Exception as e:
                    logger.info(f"WebSocket error: {e}")
                    await asyncio.sleep(2)  # 等待2秒后重连
                    continue
        except asyncio.CancelledError:
            pass
        except BinanceWebsocketUnableToConnect as e:
            logger.info(f"捕获到 Binance WebSocket 连接异常: {e}")



# 写一个消费redis队列的方法
async def consume_crypto_stream(redis_conn):
    global PREVIUS_DATA
    PREVIUS_DATA = None
    try:
        while True:
            # 1. 从最早消息开始读取（ID格式为0-0）
            # 使用XRANGE分页读取，避免单次加载过大
            messages = await redis_conn.xrange(
                config.REDIS_KEY_NAMES['all_price_stream'],
                min='0-0',  # 起始ID（最早消息）
                max='+',  # 结束ID（最新消息）
                count=3  # 每页读取3条（根据业务调整）
            )
            if not messages:
                # 无新消息时休眠1秒
                await asyncio.sleep(1)
                continue
            for msg_id, data in messages:
                # 解析消息数据（根据实际结构解析）
                data_dict =json.loads(data['data'])
                # 整分钟数据存储起来分析
                timestamp = int(float(msg_id[:-2]) / 1000)
                if datetime.fromtimestamp(timestamp).second == 0:  # 末尾截取，-2表示倒数第二个字符的前一个位置
                    # 增加时间戳字段到每一行中
                    row = {"timestamp": timestamp}
                    data_dict.update(row)
                    await redis_conn.lpush(config.REDIS_KEY_NAMES['all_price_every_time'], json.dumps(data_dict))
                    if await redis_conn.llen(config.REDIS_KEY_NAMES['all_price_every_time']) > 1 * 60 * 24:
                        await redis_conn.ltrim(config.REDIS_KEY_NAMES['all_price_every_time'], 0, 1 * 60 * 24 - 1)
                # 价格变动（突破/跌破）监测---当前价格和上一个价格做对比,
                if PREVIUS_DATA is not None:
                    watch_dict = await redis_conn.hgetall(config.REDIS_KEY_NAMES['watch_list'])
                    if len(watch_dict.items()) == 0:
                        watch_dict = {'BTC': 2, 'SOL': 2, 'GRASS': 3}
                    for s, t in watch_dict.items():
                        current_data = data_dict
                        current_price = float(current_data[f'{s}USDT'][0])
                        previus_price = float(PREVIUS_DATA[f'{s}USDT'][0])
                        current_truncate_price = truncate_high_digits(current_price, int(t))
                        previus_truncate_price = truncate_high_digits(previus_price, int(t))
                        if current_truncate_price > previus_truncate_price:
                            await send_gotify(config.GOTIFY_TOKENS['price_watcher'], message=f'现价{current_price:.3f}',
                                              title=f'🟢{s}突破{current_truncate_price}')
                        elif current_truncate_price < previus_truncate_price:
                            await send_gotify(config.GOTIFY_TOKENS['price_watcher'], message=f'现价{current_price:.3f}',
                                              title=f'🔴{s}跌破{previus_truncate_price}')
                await redis_conn.xdel(config.REDIS_KEY_NAMES['all_price_stream'], msg_id)
                PREVIUS_DATA = data_dict
    except asyncio.CancelledError:
        pass

@app.get(path="/")
def read_root():
    return {"message": "hello world"}


def truncate_high_digits(number, digits):
    """
    保留高位截取数字的方法
    :param number: 输入数字（整数或小数）
    :param digits: 要保留的高位位数
    :return: 截断后的数字（整数或浮点数）
    """
    # 处理特殊值
    if number == 0:
        return 0

    # 处理符号
    sign = '-' if number < 0 else ''
    num_str = str(abs(number))

    # 分割整数和小数部分
    if '.' in num_str:
        integer_part, fractional_part = num_str.split('.')
    else:
        integer_part, fractional_part = num_str, ''

    # 合并所有有效数字（忽略前导零）
    full_digits = integer_part.lstrip('0') + fractional_part

    # 处理全零的情况（如 0.000）
    if not full_digits:
        return 0

    # 计算实际需要保留的位数
    keep_digits = min(digits, len(full_digits))
    truncated = full_digits[:keep_digits].ljust(keep_digits, '0')

    # 重新构造数字字符串
    if len(integer_part) > 0:
        total_digits = len(integer_part) + len(fractional_part)
        if keep_digits <= len(integer_part.lstrip('0')):
            # 整数部分足够长
            new_integer = truncated + '0' * (len(integer_part) - keep_digits)
            result_str = f"{sign}{new_integer}"
        else:
            # 需要分割整数和小数部分
            overflow = keep_digits - len(integer_part.lstrip('0'))
            new_int_part = truncated[:len(integer_part)]
            new_frac_part = truncated[len(integer_part):]
            result_str = f"{sign}{new_int_part}.{new_frac_part}"
    else:
        # 纯小数情况（如 0.00123）
        result_str = f"{sign}0.{'0' * keep_digits}"
        result_str = result_str[:2 + keep_digits]  # 截断多余位数

    # 转换为数值类型并优化显示
    result = float(result_str)
    return int(result) if result.is_integer() else result

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)
